package com.dap.dubbo.rpc.utils;

import org.apache.commons.lang.StringUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.ACLBackgroundPathAndBytesable;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ZkManager {
    private static final Logger LOGGER = LoggerFactory.getLogger(ZkManager.class);
    /**
     * 客户端
     */
    private CuratorFramework client;
    /**
     * 线程池
     */
    private final ExecutorService pool = Executors.newFixedThreadPool(10);

    public ZkManager(String host) {
        //连接
        connect(host);
    }

    public void connect(String host) {
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
        this.client = connect(host, retryPolicy, Integer.MAX_VALUE, Integer.MAX_VALUE);
    }

    /**
     * 创建客户端
     * @param host
     * @param retryPolicy
     * @param connectionTimeoutMs
     * @param sessionTimeoutMs
     * @return
     */
    private CuratorFramework connect(String host, RetryPolicy retryPolicy, int connectionTimeoutMs,
                                     int sessionTimeoutMs) {
        if ((this.client == null) || (!CuratorFrameworkState.STARTED.equals(this.client.getState()))) {
            this.client =
                    CuratorFrameworkFactory.newClient(host, sessionTimeoutMs, connectionTimeoutMs,
                            retryPolicy);

            this.client.start();
        }
        return this.client;
    }

    /**
     * 检查客户端
     */
    private void checkClient() {
        if ((this.client == null) || (!CuratorFrameworkState.STARTED.equals(this.client.getState()))) {
            throw new IllegalStateException("Please connect first!");
        }
    }

    /**
     * 创建节点
     * @param path
     * @param node
     * @param isPersistent
     * @return
     */
    public boolean create(String path, String node, boolean isPersistent) {
        checkClient();
        ACLBackgroundPathAndBytesable<String> b = null;
        if (isPersistent) {
            b = this.client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT);
        } else {
            b = this.client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL);
        }
        try {
            if (!StringUtils.isEmpty(node)) {
                b.forPath(path, node.getBytes("utf-8"));
            } else {
                b.forPath(path);
            }
            return true;
        } catch (Exception e) {
            LOGGER.error(e.toString(), e);
        }
        return false;
    }

    /**
     * 更新节点
     * @param path
     * @param node
     * @return
     */
    public boolean update(String path, String node) {
        checkClient();
        try {
            this.client.setData().forPath(path,node.getBytes("utf-8"));
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     *监听本节点
     * @param path
     * @param listener
     * @return
     */
    public boolean watch(String path, ZkNodeListener listener) {
        checkClient();
        final String key = path;
        final ZkNodeListener cb = listener;
        final NodeCache nodeCache = new NodeCache(this.client, path, false);
        try {
            nodeCache.start(true);
            nodeCache.getListenable().addListener(new NodeCacheListener() {
                @Override
                public void nodeChanged() throws Exception {
                    LOGGER.debug("config [{}] changed, new data is: [{}]", key, new String(nodeCache
                            .getCurrentData().getData()));
                    cb.update(key,new String(nodeCache.getCurrentData().getData(),"UTF-8"));
                }
            }, this.pool);

            return true;
        } catch (Exception e) {
            LOGGER.error(e.toString(), e);
            try {
                nodeCache.close();
            } catch (IOException ex) {
            }
        } finally {
            Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
                public void run() {
                    try {
                        nodeCache.close();
                    } catch (IOException e) {
                    }
                }
            }));
        }
        return false;
    }

    /**
     * 监听子节点
     * @param path
     * @param listener
     * @return
     * @throws Exception
     */
    public boolean childrenWatch(String path, ZkNodeListener listener) throws Exception {
        checkClient();
        final String key = path;
        final ZkNodeListener cb = listener;
        final PathChildrenCache childrenCache = new PathChildrenCache(this.client, path, true);
        try {
            childrenCache.start(PathChildrenCache.StartMode.NORMAL);
            childrenCache.getListenable().addListener(new  PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                    ChildData data = event.getData();
                    switch (event.getType()) {
                        case CHILD_ADDED:
                            cb.updateChildren(key);
                            break;
                        case CHILD_REMOVED:
                            cb.updateChildren(key);
                            break;
                        case CHILD_UPDATED:
                            cb.updateChildren(key);
                            break;
                        default:
                            break;
                    }
                }
            }, this.pool);
            return true;
        } catch (Exception e) {
            LOGGER.error(e.toString(), e);
            try {
                childrenCache.close();
            } catch (IOException ex) {
            }
        } finally {
            Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
                public void run() {
                    try {
                        childrenCache.close();
                    } catch (IOException e) {
                    }
                }
            }));
        }
        return false;
    }

    public  String getDataNode(String path) throws Exception {
        checkClient();
        byte[] datas = client.getData().forPath(path);
        String result = new String(datas, "utf-8");
        return result;
    }

    public  List<String> getChildrenNode(String path) throws Exception {
        checkClient();
        List<String> childrenList = client.getChildren().forPath(path);
        return childrenList;
    }


    /**
     * 是否存在目录
     * @param path
     * @return
     */
    public boolean isExists(String path) {
        checkClient();
        try {
            return this.client.checkExists().forPath(path) != null;
        } catch (Exception e) {
            LOGGER.error(e.toString(), e);
        }
        return false;
    }


}
